package com.xctech.yace.service;

import com.xctech.yace.pool.ProduceThreadPool;
import com.xctech.yace.thread.ProducerThread;
import com.xctech.yace.thread.ProducerssThread;
import com.xctech.yace.util.DirUtil;
import com.xctech.yace.util.TimeUtil;
import lombok.SneakyThrows;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.io.FileInputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;


public class KafkaSend {

    private Properties properties;

    private KafkaProducer<String,String> producer;

    public KafkaSend() {
        String filePath = DirUtil.getOrgPath()+"conf/kafka.properties";
        properties = new Properties();
        try {
            properties.load(new FileInputStream(filePath));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 异步发送kafka消息，并等待返回是否发送成功
     * @param topic 主题
     * @param msg  JSON
     * @return
     */
    public boolean sendMsgAync(String topic, String msg){
        //读取kafka的配置文件
        producer = new KafkaProducer<>(properties);
        Long microTime = TimeUtil.getMicroTime();
        //发送kafka消息
        ProduceThreadPool.getInstance().submit(new ProducerThread(producer,
                new ProducerRecord<>(topic, String.valueOf(microTime), TimeUtil.resetJsTime(msg,String.valueOf(microTime)))));
        return true;
    }

    /**
     * 异步发送kafka消息，一个事务
     * @param topic 主题
     * @param msgs
     * @return
     */
    public boolean sendMsgAync(String topic, String[] msgs){
        //读取kafka的配置文件
        producer = new KafkaProducer<>(properties);
        Long microTime = TimeUtil.getMicroTime();
        //发送kafka消息
        List<ProducerRecord<String,String>> producerRecords = new ArrayList<>();
        for(int i =0;i < msgs.length;i++){
            producerRecords.add(
                    new ProducerRecord(topic, String.valueOf(microTime), TimeUtil.resetJsTime(msgs[i], String.valueOf(microTime)))
            );
        }
        ProduceThreadPool.getInstance().submit(new ProducerssThread(producer,
                producerRecords));

        return true;
    }


    /**
     * 串行发送发送kafka消息，并等待返回是否发送成功
     * @param topic 主题
     * @param msg  JSON
     * @return
     */
    @SneakyThrows
    public boolean sendMsgSync(String topic, String msg){
        Boolean isSendOk = false;
        //同步发送kafka消息
        producer = new KafkaProducer<>(properties);
        Long microTime = TimeUtil.getMicroTime();
        ProducerRecord<String, String> record = new ProducerRecord<>(topic,String.valueOf(microTime), TimeUtil.resetJsTime(msg,String.valueOf(microTime)));

        Future<RecordMetadata> future = producer.send(record);
        RecordMetadata recordMetadata = future.get();
        if(recordMetadata != null){
            isSendOk = true;
            System.out.println("发送成功,offset:" + recordMetadata.offset() + "-" + "partition:" + recordMetadata.partition());
        }
        return isSendOk;
    }

    /**
     * 串行发送发送kafka消息，不等待
     * @param topic 主题
     * @param msg  JSON
     * @return
     */
    @SneakyThrows
    public boolean sendMsgSyncNoAck(String topic, String msg){
        //同步发送kafka消息
        producer = new KafkaProducer<>(properties);
        Long microTime = TimeUtil.getMicroTime();
        producer.send(new ProducerRecord<>(topic,String.valueOf(microTime), TimeUtil.resetJsTime(msg,String.valueOf(microTime))));
        return true;
    }

}
